/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.tabletserver.action;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.client.Row;
import com.nhncorp.neptune.common.LockServiceFactory;
import com.nhncorp.neptune.common.exception.TestCaseTabletServerKillException;
import com.nhncorp.neptune.fs.GPath;
import com.nhncorp.neptune.fs.NeptuneFileSystem;
import com.nhncorp.neptune.tablet.TableSchema;
import com.nhncorp.neptune.tablet.TabletInfo;
import com.nhncorp.neptune.tabletserver.DiskSSTableIF;
import com.nhncorp.neptune.tabletserver.MemorySSTableIF;
import com.nhncorp.neptune.tabletserver.Tablet;
import com.nhncorp.neptune.tabletserver.TabletMapFile;
import com.nhncorp.neptune.tabletserver.TabletServer;
import com.nhncorp.pleiades.client.LockService;

public class TabletSplitAction extends TabletAction {
  public static final Log LOG = LogFactory.getLog(TabletSplitAction.class.getName());
  public static final String ACTION_TYPE = "TabletSplitAction";
  
  static List<String> checkActionTypes = new ArrayList<String>();
  static List<String> checkWaitingActionTypes = new ArrayList<String>();
  static List<String> checkActionTypesForFinish = new ArrayList<String>();

  static {  
    checkActionTypes.add(MinorCompactionAction.ACTION_TYPE);
    checkActionTypes.add(MajorCompactionAction.ACTION_TYPE);
    checkActionTypes.add(TabletSplitAction.ACTION_TYPE);
    checkActionTypes.add(TabletSplitFinishAction.ACTION_TYPE);
    checkActionTypes.add(TabletDropAction.ACTION_TYPE);
    checkActionTypes.add(BatchUploadAction.ACTION_TYPE);
    //checkActionTypes.add(ScannerAction.ACTION_TYPE);
    checkActionTypes.add(TabletLockAction.ACTION_TYPE);
    
    checkWaitingActionTypes.add(TabletSplitFinishAction.ACTION_TYPE);

    checkActionTypesForFinish.add(ScannerAction.ACTION_TYPE);
    checkActionTypesForFinish.add(CommitAction.ACTION_TYPE);
    checkActionTypesForFinish.add(ApplyAction.ACTION_TYPE);
    checkActionTypesForFinish.add(TabletDropAction.ACTION_TYPE);
    checkActionTypesForFinish.add(TxAction.ACTION_TYPE);
    checkActionTypesForFinish.add(TabletLockAction.ACTION_TYPE);
  }
  
  private Tablet tablet;
  private TabletServer tabletServer;
  
  private Tablet[] splitedTablets;
  private TabletInfo[] splitedTabletInfos;
  private DiskSSTableIF[] splitedDiskSSTables;
  private Row.Key midRowKey;
  private boolean end = true;
  private NeptuneFileSystem fs; 
  private Object monitor = new Object();
  
  private boolean singleRowTablet = false;
  
  public TabletSplitAction(Tablet tablet, TabletServer tabletServer) {
    this.tablet = tablet;
    this.tabletServer = tabletServer;
    if(tablet != null) {
      this.fs = NeptuneFileSystem.get(tablet.getConf());
    }
  }
  
  public boolean isThreadMode() {
    return true;
  }

  public boolean isWaitingMode() {
    return true;
  }

  public void run() {
    synchronized(monitor) {
      if(singleRowTablet) {
        return;
      }
      ////////////////////////////////////
      //for TestCase
      if(tablet == null || !tablet.isReady() || tablet.isDroped()) {
        tablet.getActionChecker().endAction(this);
        end = true;       
        return;
      }
      ////////////////////////////////////
  
      long startTime = System.currentTimeMillis();
      TabletInfo tabletInfo = tablet.getTabletInfo();
      LOG.debug("TabletSplitAction: start: " + tabletInfo);
      
      try {
        //TabletSplitAction 생성전에 Tablet에서 먼저 확인했지만
        //wait된 경우 다른 Map 파일이 만들어 졌을수도 있기 때문에 다시 한번 확인한다.
        //if(tablet.getMapFileCount() != 1 || tablet.sumMapFileSize() < maxTabletSizeByte) {
//        if(tablet.getMapFileCount() != 1) {
//          //LOG.debug("TabletSpliter: split stoped. cause tablet.getMapFileCount() > 1:" + tablet.getTabletInfo());
//          //return;
//          LOG.debug("TabletSpliter: tablet.getMapFileCount() > 1. do merge:" + tablet.getTabletInfo());
//          tablet.getDiskSSTable().majorCompaction();
//        }      
        //LOG.debug("TabletSplitAction2 : " + tabletInfo.getTabletName());
        this.splitedTabletInfos = makeSplitTabletInfo(tabletInfo);
        if(splitedTabletInfos == null) {
          singleRowTablet = true;
          LOG.warn("Can't find mid rowkey:" + tabletInfo);
          return;
        }
        //LOG.debug("TabletSplitAction3 : " + tabletInfo.getTabletName());
        if(tabletInfo.getEndRowKey().equals(splitedTabletInfos[0].getEndRowKey()) ||
            splitedTabletInfos[0].getEndRowKey().equals(splitedTabletInfos[1])) {

          singleRowTablet = true;
          
          LOG.warn("Splited Tablet has same end rowkey:" + tabletInfo + "," + 
              splitedTabletInfos[0] + "," +
              splitedTabletInfos[1]);
          return;
        }
        //LOG.debug("TabletSplitAction3 : " + tabletInfo.getTabletName());
  
        if(splitedTabletInfos == null) {
          LOG.error("Error while spliting:" + tabletInfo + ", can't find mid row key");
          return;
          //System.exit(0);
        }
        //여기까지 처리에서 에러가 발생한 경우에는 아무런 처리를 해줄 필요 없다.
        
        //DiskSSTable을 먼저 두개로 split 한다.
        this.splitedDiskSSTables = tablet.getDiskSSTable().split(midRowKey, splitedTabletInfos);
        if(this.splitedDiskSSTables == null) {
          LOG.warn("splited disk sstables are null");
          return;
        }
        
        this.splitedTablets = new Tablet[2];
        
        //TestCase에서는 TabletServer가 Null
        String hostName = tabletServer != null ? tabletServer.getHostName() : "TestCase";
        
        LockService lockService = LockServiceFactory.getLockService(tabletServer.getConf(), hostName, true);
        if(lockService == null) {
          LOG.warn("lock service is null");
          return;
        }
  
        TableSchema table = TableSchema.loadTableSchema(tablet.getConf(), lockService, tabletInfo.getTableName());
        for(int i = 0; i < 2; i++) {
          splitedTablets[i] = new Tablet(tablet.getConf(), table, tabletServer, splitedTabletInfos[i]);
          splitedTablets[i].setDiskSSTable(splitedDiskSSTables[i]);
        }
        
        try {
          Map<String, TabletMapFile[]> savedMapFiles = tablet.getMemorySSTable().splitAndSaveCanCommit(midRowKey, splitedTabletInfos);
          addMapFileToDiskSSTable(savedMapFiles);
        } catch (IOException e) {
          tablet.getMemorySSTable().cancelCompaction(null);
          throw e;
        }
        
        long endTime = System.currentTimeMillis();
        LOG.debug("SplitAction time=" + (endTime - startTime) + ", call TabletSplitFinishAction: " + tabletInfo + "," + 
            splitedTabletInfos[0] + ", " + splitedTabletInfos[1]);
        //파일의 분리 작업이 완료되면 메모리의 내용도 분리한다.
        //메모리의 내용을 분리하기 전에 Tablet에서 다른 처리를 하고 있으면 기다린다.(따라서 다른 Action으로 분리하여 처리)
        TabletSplitFinishAction tabletSplitFinishAction = new TabletSplitFinishAction(this);
        tablet.getActionChecker().startAction(tabletSplitFinishAction);
      } catch (Exception e) {
        LOG.error("split run error:" + e.getMessage(), e);
        //장애 발생시 Temp 내용 삭제
        rollback();
      } finally {
        tablet.getActionChecker().endAction(this);
        end = true;         
      }
    }
  }

  private void rollback() {
    if(this.splitedTabletInfos != null) {
      for(int i = 0; i < this.splitedTabletInfos.length; i++) {
        try {
//          GPath tempPath = new GPath(tablet.getConf().get("neptune.root") + "/temp/split/" + 
//              this.splitedTabletInfos[i].getTableName() + "/" + 
//              this.splitedTabletInfos[i].getTabletName());
          GPath tempPath = Tablet.getTabletSplitTempPath(tablet.getConf(), splitedTabletInfos[i]);
          fs.delete(tempPath, true);
          LOG.error("split run error: clear split temp dir:" + tempPath);
        } catch (Exception e) {
          LOG.error("Error while split rollback:" + e.getMessage(), e);
        }
      }
    }
  }
  private TabletInfo[] makeSplitTabletInfo(TabletInfo tabletInfo) throws IOException {
    TabletInfo[] splitedTabletInfos = new TabletInfo[2];
    splitedTabletInfos[0] = new TabletInfo(tabletInfo.getTableName(), Tablet.generateTabletName(tabletInfo.getTableName()));
    splitedTabletInfos[1] = new TabletInfo(tabletInfo.getTableName(), Tablet.generateTabletName(tabletInfo.getTableName()));
    
    //첫번째 Tablet은 다른 Tablet으로 할당, 두번째 Tablet은 자신이 서비스
    splitedTabletInfos[1].setAssignedHostName(tabletInfo.getAssignedHostName());
    
    this.midRowKey = tablet.getDiskSSTable().findMidRowKeyForSplit();
    //midRow.Key가 null인 경우는 diskSSTable에 MapFile이 하나 이상인 경우
    //이 경우에는 다음 MajorCompaction 수행 후 처리된다.
    if(midRowKey == null) {
      midRowKey = tablet.getMemorySSTable().findMidRowKeyForSplit();
      if(midRowKey == null) {
        return null;
      }
    }
    //splitedTabletInfos[0].setStartRow.Key(tabletInfo.getStartRowKey(());
    splitedTabletInfos[0].setEndRowKey(midRowKey);
    splitedTabletInfos[1].setEndRowKey(tabletInfo.getEndRowKey());
    return splitedTabletInfos;
  }
  
  public void setEnd(boolean end) {
    this.end = end;
  }
  
  public boolean isEnd() {
    return end;
  }
  
  public Tablet[] getSplitedTablets() {
    return splitedTablets;
  }
  
  class TabletSplitFinishAction extends TabletAction { 
    public static final String ACTION_TYPE = "TabletSplitFinishAction";
    
    private TabletSplitAction tabletSplitAction;
    
    public TabletSplitFinishAction(TabletSplitAction tabletSplitAction) {
      this.tabletSplitAction = tabletSplitAction;
    }
    
    public boolean isThreadMode() {
      return true;
    }

    public boolean isWaitingMode() {
      return true;
    }

    public void run() {
      MemorySSTableIF memorySSTable = tablet.getMemorySSTable();
      try {
        long startTime = System.currentTimeMillis();
        if(tablet == null || !tablet.isReady() || tablet.isDroped()) {
          if(tablet == null) {
            LOG.debug("TabletSplitFinishAction: stoped: tablet is null");
          } else {
            LOG.debug("TabletSplitFinishAction: stoped:ready=" + tablet.isReady() + ", droped=" + tablet.isDroped());
          }
          return;
        }
        TabletInfo tabletInfo = tablet.getTabletInfo();
        LOG.debug("TabletSplitFinishAction: start: " + tabletInfo);

        //memory에 있는 데이터 Split
        Map<String, TabletMapFile[]> savedMapFiles = null;
        try {
          savedMapFiles = memorySSTable.splitAndSave(midRowKey, splitedTabletInfos);
          addMapFileToDiskSSTable(savedMapFiles);
          //LOG.debug("Memory Splited");
        } catch (TestCaseTabletServerKillException e) {
          //Test예외 Exception이 오면 바로 종료
          return;
        } catch (Exception e) {
          //오류 발생시 temp dir 삭제
          LOG.error("MemorySSTable.splitAndSave", e);
          rollback();
          memorySSTable.cancelCompaction(null);
          return;
        }
  
        //Tablet 파일 Move
        try {
          splitedTablets[0].getDiskSSTable().moveTempToTablet(false);
          splitedTablets[1].getDiskSSTable().moveTempToTablet(false);
        } catch (TestCaseTabletServerKillException e) {
          //Test예외 Exception이 오면 바로 종료
          return;
        } catch (Exception e) {
          //오류 발생 시 temp dir 삭제
          LOG.error("moveTempToTablet error:" + tabletInfo, e);
          rollback();
          memorySSTable.cancelCompaction(null);
          
          //move된 tablet 파일 삭제
          for(int i = 0; i < 2; i++) {
            try {
              GPath dataPath = Tablet.getTabletPath(tablet.getConf(), splitedTablets[i].getTabletInfo());
              fs.delete(dataPath, true);
            } catch (Exception e1) {
              e1.printStackTrace();
            }
          }
          return;
        }

        //TabletServer에 등록 & Meta Data에 등록
        try {
          if(tabletServer != null) {
            tabletServer.tabletSplited(tablet, splitedTablets);
          }
          LOG.debug("TabletSplitFinishAction: after tabletSplited: " + tabletInfo);
          tablet.setReady(false);
        } catch (TestCaseTabletServerKillException e) {
          //Test예외 Exception이 오면(TabletServer kill 상황) 바로 종료
          LOG.error("Stoped Tablet Split", e);
          return;
        } catch (Exception e) {
          //META 데이터 복구는 tabletServer의 tabletSplited에서 수행 -> 복구 필요 없음(3개의 작업을 1tx에서 처리)
          //temp dir 삭제
          LOG.error("modify meta info error", e);
          rollback(); 
          tablet.getMemorySSTable().cancelCompaction(null);
          //move된 tablet 파일 삭제
          for(int i = 0; i < 2; i++) {
            try {
              GPath dataPath = Tablet.getTabletPath(tablet.getConf(), splitedTablets[i].getTabletInfo());
              fs.delete(dataPath, true);
            } catch (Exception e1) {
              e1.printStackTrace();
            }
          }   
          
          return;
        }
        
        try {
          //기존 Tablet Data 삭제
          tablet.deleteTablet();
        } catch (Exception e) {
          //여기서 오류가 발생해도 아무런 처리를 하지 않아도 됨
          //TODO 나중에 별도로 전체적으로 clean 작업해주는 필요 
          LOG.error("old tablet drop error", e);
        }
        
        //temp file 삭제
        try {
          splitedTablets[0].deleteSplitTempDir();
          splitedTablets[1].deleteSplitTempDir();
        } catch (Exception e) {
          //여기서 오류가 발생해도 아무런 처리를 하지 않아도 됨
          //TODO 나중에 별도로 전체적으로 clean 작업해주는 필요 
          LOG.error("deleteSplitTempDir", e);
        }
        
        //split된 tablet 서비스 준비 완료 설정
        //splitedTablets[0].setReady(true);
        splitedTablets[1].setReady(true);

        LOG.debug("TabletSplitFinishAction: End: " + tabletInfo + "," + 
            splitedTabletInfos[0] + ", " + splitedTabletInfos[1] + ", time=" + (System.currentTimeMillis() - startTime));
        //LOG.debug("[" + splitedTabletInfos[0] + "][" + splitedTabletInfos[1] + "]");
      } catch (Exception e) {
        LOG.error("run error", e);
      } finally {
        tablet.getActionChecker().endAction(this);
        tablet.getActionChecker().endAction(tabletSplitAction);
        end = true;
      }
    }

    @Override
    public String getActionKey() {
      return tablet == null ? "null:" + getActionType() : tablet.getTabletInfo().getTabletName() + getActionType();
      //return getActionType();
    }

    @Override
    public String getActionType() {
      return ACTION_TYPE;
    }

    @Override
    public List<String> getRunCheckActionTypes() {
      return checkActionTypesForFinish;
    }
  }

  private void addMapFileToDiskSSTable(
      Map<String, TabletMapFile[]> savedMapFiles) throws IOException {
    for(Map.Entry<String, TabletMapFile[]> entry: savedMapFiles.entrySet()) {
      String columnName = entry.getKey();
      TabletMapFile[] tabletMapFiles = entry.getValue();
      for(int i = 0; i < 2; i++) {
        //LOG.debug("addTabletMapFile:tabletName=" + splitedTablets[i].getTabletInfo().getTabletName() + ", path=" + tabletMapFiles[i].getFilePath() + ", size=" + tabletMapFiles[i].getDataFileSize());
        if(tabletMapFiles[0] != null) {
          //GPath targetPath = new GPath(Tablet.getTabletPath(tablet.getConf(), splitedTabletInfos[i]), columnName + "/" + tabletMapFiles[i].getFileId() + "/");
          //tabletMapFiles[i].moveTo(targetPath, false);
          splitedTablets[i].getDiskSSTable().addTabletMapFile(columnName, tabletMapFiles[i]);              
        }
      }
    }
  }
  
  @Override
  public String getActionKey() {
    return tablet == null ? "null:" + getActionType() : tablet.getTabletInfo().getTabletName() + getActionType();

    //return getActionType();
  }

  public TabletSplitFinishAction getTabletSplitFinishAction() {
    return new TabletSplitFinishAction(this);
  }

  @Override
  public String getActionType() {
    return ACTION_TYPE;
  }

  @Override
  public List<String> getRunCheckActionTypes() {
    return checkActionTypes;
  }
  
  public List<String> getWaitCheckActionTypes() {
    return checkWaitingActionTypes;
  }
}
